package com.pactera.asmp.server.common.websocket;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pactera.asmp.server.common.lock.Lock;
import com.pactera.asmp.server.common.logprint.utils.CollectionUtils;
import com.pactera.asmp.server.common.utils.SpringUtil;
import com.pactera.asmp.server.pojo.User;
import com.pactera.asmp.server.pojo.WSMessage;
import com.pactera.asmp.server.service.IDeviceServcie;
import com.pactera.asmp.server.service.UserService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

@ServerEndpoint("/ws/webcan/v1/{userid}/{duid}")
@Component
public class WebSocketServerWebCan {
    public static final String CODE_DATA_CAN_START = "914"; // 发送CAN信息
    public static final String CODE_DATA_CAN_STOP = "915"; // 停止发送CAN信息
    private Logger logger = LoggerFactory.getLogger(WebSocketServerWebCan.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    private Lock lock;
    private SocketUtil socketUtil;
    private StringRedisTemplate stringRedisTemplate;
    private IDeviceServcie deviceServcie;
    private UserService userService;

    public static volatile ConcurrentHashMap<String, Session> id2sessionMap = new ConcurrentHashMap<>();
    public static volatile ConcurrentHashMap<Session, String> session2idMap = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam("userid") String userid, @PathParam("duid") String duid) throws IOException {
        if(StringUtils.isEmpty(userid) || StringUtils.isEmpty(duid)){
            logger.info("参数异常！！！");
            session.close();
            return;
        }
        User userById = userService().findUserById(Integer.valueOf(userid));
        if(userById == null || userById.getIsDel()){
            logger.info("查无有效用户");
            session.close();
            return;
        }
        String sessionKey = socketUtil().formatSessionId(userid,session);
        String username;
        username = userById.getUsername();
        if(StringUtils.isEmpty(sessionKey)){
            session.close();
            return;
        }
        sessionKey = "CAN" +sessionKey;

        lockService().lock(userid);
        try {
            logger.info("CanUser onOpen =================================================================================================================={}", sessionKey);
            session.setMaxIdleTimeout(15000);  // 15秒超时过期

            // 1. 本地缓存
            id2sessionMap.put(sessionKey, session);
            session2idMap.put(session, sessionKey);
            // 2. 分布式Session服务注册
            socketUtil().registerSocketServer(sessionKey);
            // 3. 心跳信息
            socketUtil().heartBeatById(sessionKey);

//            logger.info("======================={},{},{}",duid,userid,username);
            // 设备占用
            boolean bindSuccess = deviceServcie().operationDevice(duid, Integer.valueOf(userid), 1, username);
//            logger.info("======================={}",bindSuccess);
            if(!bindSuccess){
                session.close();
                return;
            }

            //发送消息给设备，告诉发送CAN相关数据
            // 发送MQ消息, (默认不发送，前端手动调用来开启发送CAN信息**)
//            WSMessage message = new WSMessage();
//            message.setMsgcode(CODE_DATA_CAN_START);
//            message.setDescribe("发送CAN信息");
//            socketUtil().sendMQ(userid,duid, objectMapper.writeValueAsBytes(message), "json",null);

            // 5. 成功连接反馈
            socketUtil().openSuccessMessage(session, userid,sessionKey);
        }
        catch (Exception e){
            e.printStackTrace();
            logger.error("Can OnOpen Error: {}",e.getMessage());
        }
        finally {
            lockService().unlock(userid);
        }
    }


    private Integer getDeviceStatus(String duid){
        Set<String> members = stringRedisTemplate().opsForSet().members(SocketUtil.REDIS_KEY_WS_DEVICE_INSTANCE + ":" + duid);
        if(!CollectionUtils.isEmpty(members)){
            Long rank;
            for(String instance : members){
                rank = stringRedisTemplate().opsForZSet().rank(HeartBeatService.REDIS_KEY_WS_ONLINE_MEMBERS, instance);
                if(rank != null && rank.intValue() > -1){
                  return 0;  // 设备在线,设置为空闲
                }
            }
        }
        return 2; // 设备断线
    }

    @OnClose
    public void onClose(Session session) {
        if (session2idMap.containsKey(session)) {
            String id = session2idMap.get(session);
            String canuserid = id.split("@")[0];
            String userid = canuserid.substring(3);
            lockService().lock(userid);
            try {
                logger.info("CanUser onClose =================================================================================================================={},{}", id,userid);
                id2sessionMap.remove(id);
                session2idMap.remove(session);

                // 终端设备状态修改  空闲/断线
                String duid = stringRedisTemplate().opsForValue().get(userid);
                if(!StringUtils.isEmpty(duid)) {
                    Integer duidStatus = this.getDeviceStatus(duid);
                    deviceServcie().updateStatus(duid, duidStatus);
                    // 发送MQ消息
                    WSMessage message = new WSMessage();
                    message.setMsgcode(CODE_DATA_CAN_STOP);
                    message.setDescribe("停止发送CAN信息");
                    socketUtil().sendMQ(userid,duid, objectMapper.writeValueAsBytes(message), "json",null);
                }

                // 解绑关联
                try {
                    stringRedisTemplate().delete(duid);  // 真机一对一关联
                    stringRedisTemplate().delete(userid);  // 真机一对一关联
                }catch (Exception ex){
                    ex.printStackTrace();
                    logger.info(".................{}",ex.getMessage());
                }
                stringRedisTemplate().opsForHash().delete(SocketUtil.REDIS_KEY_SOCKET_SERVER_USER,id);  //  SERVER 关联
                stringRedisTemplate().opsForSet().remove(SocketUtil.REDIS_KEY_WS_USER_INSTANCE+":"+canuserid,id); // 用户ID和实例ID的关联
                socketUtil().offlineRedisCache(id);
            } catch (Exception e) {
                e.printStackTrace();
                logger.info(e.getMessage());
            } finally {
                lockService().unlock(userid);
            }
        }
    }

    /**
     *
     * @param message
     * 1. ping#XXXXXX     心跳续约
     * 2. status:open     发送CAN信息
     * 3. status:stop     停止发送CAN信息
     * @param session
     * @throws IOException
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        if(!StringUtils.isEmpty(message) && message.startsWith("ping#")){
            String id = session2idMap.get(session);
            message = "ping#"+id;
            socketUtil().heartBeatByMsg(message);
            return;
        }
        String id = session2idMap.get(session);
        String userId = id.split("@")[0].substring(3);
        String duid = stringRedisTemplate().opsForValue().get(userId);
        if(!StringUtils.isEmpty(duid)) {
            try {
                WSMessage sockMessage = new WSMessage();
                if(message.startsWith("status:open")){
                    sockMessage.setMsgcode(CODE_DATA_CAN_START);
                    sockMessage.setDescribe("发送CAN信息");
                }
                else if(message.startsWith("status:stop")){
                    sockMessage.setMsgcode(CODE_DATA_CAN_STOP);
                    sockMessage.setDescribe("停止发送CAN信息");
                }
                else {
                    return;
                }
                socketUtil().sendMQ(userId,duid, objectMapper.writeValueAsBytes(sockMessage), "json",null);
            } catch (JsonProcessingException e) {
                logger.error("CanUser onMessage Error: {}",e.getMessage());
            }
        }
    }

    @OnMessage
    public void onMessage(byte[] bytes, Session session) {

    }

    @OnError
    public void onError(Session session, Throwable error) {
        String id = session2idMap.get(session);
        logger.error("设备连接错误:{},原因:{}", id, error.getMessage());
    }

    private Lock lockService(){
        if(this.lock == null) this.lock =  SpringUtil.getBean(Lock.class);
        return this.lock;
    }
    private SocketUtil socketUtil(){
        if(this.socketUtil == null) this.socketUtil =  SpringUtil.getBean(SocketUtil.class);
        return this.socketUtil;
    }
    private StringRedisTemplate stringRedisTemplate(){
        if(this.stringRedisTemplate == null) this.stringRedisTemplate =  SpringUtil.getBean(StringRedisTemplate.class);
        return this.stringRedisTemplate;
    }
    private IDeviceServcie deviceServcie(){
        if(this.deviceServcie == null) this.deviceServcie =  SpringUtil.getBean(IDeviceServcie.class);
        return this.deviceServcie;
    }

    private UserService userService(){
        if(this.userService == null) this.userService =  SpringUtil.getBean(UserService.class);
        return this.userService;
    }
}

